package io.mycat.migrate;

import com.alibaba.druid.util.JdbcUtils;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.google.common.base.Strings;
import io.mycat.MycatServer;
import io.mycat.backend.datasource.PhysicalDBNode;
import io.mycat.route.function.PartitionByCRC32PreSlot;
import io.mycat.server.util.SchemaUtil;
import io.mycat.sqlengine.OneRawSQLQueryResultHandler;
import io.mycat.sqlengine.SQLJob;
import io.mycat.util.DateUtil;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.Serializable;
import java.math.BigInteger;
import java.nio.charset.Charset;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.*;

import static io.mycat.util.dataMigrator.DataMigratorUtil.executeQuery;

public class BinlogStream {
    private static Logger logger = LoggerFactory.getLogger(BinlogStream.class);
    private final String hostname;
    private final int port;
    private final String username;
    private final String password;
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    private BinaryLogClient binaryLogClient;
    private Charset charset;

    private long slaveID;
    private String binglogFile;
    private long binlogPos;

    private Set<String> databaseSet = new HashSet<>();
    private Map<String, Semaphore> semaphoreMap = new ConcurrentHashMap<>();


    private List<MigrateTask> migrateTaskList;

    public List<MigrateTask> getMigrateTaskList() {
        return migrateTaskList;
    }

    public void setMigrateTaskList(List<MigrateTask> migrateTaskList) {
        this.migrateTaskList = migrateTaskList;
        for (MigrateTask migrateTask : migrateTaskList) {
            databaseSet.add(MigrateUtils.getDatabaseFromDataNode(migrateTask.getFrom()));
            String dataHostTo = MigrateUtils.getDataHostFromDataNode(migrateTask.getTo());
            if (!semaphoreMap.containsKey(dataHostTo)) {
                int count = Double.valueOf(MycatServer.getInstance().getConfig().getDataHosts().get(dataHostTo).getSource().getSize() * 0.8).intValue();
                semaphoreMap.put(dataHostTo, new Semaphore(1));
            }
        }
    }

    public long getSlaveID() {
        return slaveID;
    }

    public void setSlaveID(long slaveID) {
        this.slaveID = slaveID;
    }

    public String getBinglogFile() {
        return binglogFile;
    }

    public void setBinglogFile(String binglogFile) {
        this.binglogFile = binglogFile;
    }

    public long getBinlogPos() {
        return binlogPos;
    }

    public void setBinlogPos(long binlogPos) {
        this.binlogPos = binlogPos;
    }

    private volatile boolean groupEventsByTX = true;


    public BinlogStream(String hostname, int port, String username, String password, Charset charset) {
        this.hostname = hostname;
        this.port = port;
        this.username = username;
        this.password = password;
        this.charset = charset;
    }

    public void setGroupEventsByTX(boolean groupEventsByTX) {
        this.groupEventsByTX = groupEventsByTX;
    }


    public void connect() throws IOException {
        initTaskDate();
        scheduler.scheduleAtFixedRate(new BinlogIdleCheck(this), 5, 15, TimeUnit.SECONDS);
        allocateBinaryLogClient().connect();

    }

    private void initTaskDate() {
        Date curDate = new Date();
        for (MigrateTask migrateTask : migrateTaskList) {
            migrateTask.setLastBinlogDate(curDate);
        }
    }

    public void connect(long timeoutInMilliseconds) throws IOException, TimeoutException {
        initTaskDate();
        scheduler.scheduleAtFixedRate(new BinlogIdleCheck(this), 5, 15, TimeUnit.SECONDS);
        allocateBinaryLogClient().connect(timeoutInMilliseconds);

    }

    private synchronized BinaryLogClient allocateBinaryLogClient() {
        if (isConnected()) {
            throw new IllegalStateException("MySQL replication stream is already open");
        }
        binaryLogClient = new BinaryLogClient(hostname, port, username, password);
        binaryLogClient.setBinlogFilename(getBinglogFile());
        binaryLogClient.setBinlogPosition(getBinlogPos());
        binaryLogClient.setServerId(getSlaveID());
        binaryLogClient.registerEventListener(new DelegatingEventListener());
        return binaryLogClient;
    }


    public synchronized boolean isConnected() {
        return binaryLogClient != null && binaryLogClient.isConnected();
    }


    public synchronized void disconnect() throws IOException {
        if (binaryLogClient != null) {
            binaryLogClient.disconnect();
            binaryLogClient = null;
        }
        shutdownAndAwaitTermination(scheduler);
    }


    void shutdownAndAwaitTermination(ExecutorService pool) {
        pool.shutdown(); // Disable new tasks from being submitted
        try {
            // Wait a while for existing tasks to terminate
            if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
                pool.shutdownNow(); // Cancel currently executing tasks
                // Wait a while for tasks to respond to being cancelled
                if (!pool.awaitTermination(60, TimeUnit.SECONDS))
                    logger.warn("Pool did not terminate");
            }
        } catch (InterruptedException ie) {
            // (Re-)Cancel if current thread also interrupted
            pool.shutdownNow();
            // Preserve interrupt status
            Thread.currentThread().interrupt();
        }
    }


    private final class DelegatingEventListener implements BinaryLogClient.EventListener {

        private final Map<Long, TableMapEventData> tablesById = new HashMap<Long, TableMapEventData>();
        private final Map<String, Map<Integer, Map<String, Object>>> tablesColumnMap = new HashMap<>();

        private boolean transactionInProgress;
        private String binlogFilename;


        //当发现ddl语句时 需要更新重新取列名
        private Map<Integer, Map<String, Object>> loadColumn(String database, String table) {
            Map<Integer, Map<String, Object>> rtn = new HashMap<>();
            List<Map<String, Object>> list = null;
            Connection con = null;
            try {
                con = DriverManager.getConnection("jdbc:mysql://" + hostname + ":" + port, username, password);
                list = executeQuery(con, "select  COLUMN_NAME, ORDINAL_POSITION, DATA_TYPE, CHARACTER_SET_NAME from INFORMATION_SCHEMA.COLUMNS where table_name='" + table + "' and TABLE_SCHEMA='" + database + "'");

            } catch (SQLException e) {
                throw new RuntimeException(e);
            } finally {
                JdbcUtils.close(con);
            }
            for (Map<String, Object> stringObjectMap : list) {
                BigInteger pos = (BigInteger) stringObjectMap.get("ORDINAL_POSITION");
                rtn.put(pos.intValue(), stringObjectMap);
            }
            return rtn;
        }

        @Override
        public void onEvent(Event event) {
            logger.debug("----->migrate binlog event:" + event.toString());
            EventType eventType = event.getHeader().getEventType();
            switch (eventType) {
                case TABLE_MAP:
                    TableMapEventData tableMapEventData = event.getData();
                    tablesById.put(tableMapEventData.getTableId(), tableMapEventData);
                    if (!tablesColumnMap.containsKey(tableMapEventData.getDatabase() + "." + tableMapEventData.getTable())) {
                        tablesColumnMap.put(tableMapEventData.getDatabase() + "." + tableMapEventData.getTable(), loadColumn(tableMapEventData.getDatabase(), tableMapEventData.getTable()));
                    }
                    break;
                case ROTATE:
                    RotateEventData data = event.getData();
                    binlogFilename = data.getBinlogFilename();
                    break;
                case PRE_GA_WRITE_ROWS:
                case WRITE_ROWS:
                case EXT_WRITE_ROWS:
                    handleWriteRowsEvent(event);
                    break;
                case PRE_GA_UPDATE_ROWS:
                case UPDATE_ROWS:
                case EXT_UPDATE_ROWS:
                    handleUpdateRowsEvent(event);
                    break;
                case PRE_GA_DELETE_ROWS:
                case DELETE_ROWS:
                case EXT_DELETE_ROWS:
                    handleDeleteRowsEvent(event);
                    break;
                case QUERY:
                    if (groupEventsByTX) {
                        QueryEventData queryEventData = event.getData();
                        String query = queryEventData.getSql();
                        if ("BEGIN".equals(query)) {
                            transactionInProgress = true;
                        } else if (!query.startsWith("#")) {
                            handleOtherSqlEvent(event);
                        }
                    }
                    break;
                case XID:
                    if (groupEventsByTX) {
                        transactionInProgress = false;
                    }

                    break;
                default:
                    // ignore
            }
        }

        private void exeSql(MigrateTask task, String sql) {
            if (task.isHaserror())
                return;
            task.setHasExecute(true);
            String dataHostTo = MigrateUtils.getDataHostFromDataNode(task.getTo());
            Semaphore semaphore = semaphoreMap.get(dataHostTo);
            try {
                semaphore.acquire();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            SqlExecuteListener listener = new SqlExecuteListener(task, sql, BinlogStream.this,
                    semaphore);
            OneRawSQLQueryResultHandler resultHandler = new OneRawSQLQueryResultHandler(new String[0],
                    listener);
            resultHandler.setMark("binlog execute");
            PhysicalDBNode dn = MycatServer.getInstance().getConfig().getDataNodes().get(task.getTo());
            SQLJob sqlJob = new SQLJob(sql, dn.getDatabase(), resultHandler, dn.getDbPool().getSource());
            listener.setSqlJob(sqlJob);
            sqlJob.run();
        }

        private void handleOtherSqlEvent(Event event) {
            QueryEventData queryEventData = event.getData();
            logger.debug("receve sql:", queryEventData.getSql());
            SchemaUtil.SchemaInfo schemaInfo = SchemaUtil.parseSchema(queryEventData.getSql());
            if (isShouldBeFilter(queryEventData.getDatabase(), schemaInfo.table))
                return;
            String query = queryEventData.getSql();
            for (MigrateTask migrateTask : migrateTaskList) {
                if (schemaInfo.table.equalsIgnoreCase(migrateTask.getTable())
                        && queryEventData.getDatabase().equalsIgnoreCase(MigrateUtils.getDatabaseFromDataNode(migrateTask.getFrom()))) {
                    exeSql(migrateTask, query);
                }
            }


        }

        private boolean isShouldBeFilter(String database, String table) {
            if (Strings.isNullOrEmpty(database))
                return true;
            if (Strings.isNullOrEmpty(table))
                return true;
            if (!databaseSet.contains(database.toLowerCase())) {
                return true;
            }
            for (MigrateTask migrateTask : migrateTaskList) {
                if (database.equals(MigrateUtils.getDatabaseFromDataNode(migrateTask.getFrom())) && table.equalsIgnoreCase(migrateTask.getTable())) {
                    return false;
                }
            }


            return true;
        }


        private void handleWriteRowsEvent(Event event) {
            WriteRowsEventData eventData = event.getData();
            TableMapEventData tableMapEvent = tablesById.get(eventData.getTableId());
            if (isShouldBeFilter(tableMapEvent.getDatabase(), tableMapEvent.getTable()))
                return;
            Map<Integer, Map<String, Object>> xxx = tablesColumnMap.get(tableMapEvent.getDatabase() + "." + tableMapEvent.getTable());
            BitSet inculudeColumn = eventData.getIncludedColumns();
            StringBuilder sb = new StringBuilder("insert into  ");
            sb.append(tableMapEvent.getTable());
            sb.append("(");
            int size = inculudeColumn.length();
            List<Serializable[]> rows = eventData.getRows();

            int slot = -1;
            for (int i = 0; i < size; i++) {
                int column = inculudeColumn.nextSetBit(i);
                Map<String, Object> coumnMap = xxx.get(column + 1);
                sb.append(coumnMap.get("COLUMN_NAME"));
                if (i != size - 1) {
                    sb.append(",");
                }
            }
            sb.append(")  values  ");
            for (int i = 0; i < rows.size(); i++) {
                Serializable[] value = rows.get(i);
                sb.append(" (");
                for (int y = 0; y < size; y++) {
                    int column = inculudeColumn.nextSetBit(y);
                    Map<String, Object> coumnMap = xxx.get(column + 1);
                    String dataType = (String) coumnMap.get("DATA_TYPE");
                    String columnName = (String) coumnMap.get("COLUMN_NAME");
                    if ("_slot".equalsIgnoreCase(columnName)) {
                        slot = value[y] instanceof BigInteger ? ((BigInteger) value[y]).intValue() : ((Integer) value[y]);
                    }
                    sb.append(convertBinlogValue(value[y], dataType));

                    if (y != size - 1) {
                        sb.append(",");
                    }
                }
                sb.append(")");
                if (i != rows.size() - 1) {
                    sb.append(",");
                }
            }

            checkIfExeSql(tableMapEvent, sb, slot);

        }

        private void checkIfExeSql(TableMapEventData tableMapEvent, StringBuilder sb, int slot) {
            for (MigrateTask migrateTask : migrateTaskList) {
                if (tableMapEvent.getTable().equalsIgnoreCase(migrateTask.getTable())
                        && tableMapEvent.getDatabase().equalsIgnoreCase(MigrateUtils.getDatabaseFromDataNode(migrateTask.getFrom()))) {
                    for (PartitionByCRC32PreSlot.Range range : migrateTask.getSlots()) {
                        if (range.end >= slot && range.start <= slot) {
                            exeSql(migrateTask, sb.toString());
                            return;
                        }
                    }

                }
            }
        }

        private Object convertBinlogValue(Serializable value, String dataType) {
            if (value instanceof String) {
                return "'" + ((String) value).replace("'", "\\'") + "'";
            } else if (value instanceof byte[]) {
                //确认编码
                return "'" + new String((byte[]) value, charset).replace("'", "\\'") + "'";
            } else if (value instanceof Date) {
                return "'" + dateToString((Date) value, dataType) + "'";
            } else if (("date".equalsIgnoreCase(dataType)) && value instanceof Long) {
                return "'" + dateToStringFromUTC((Long) value) + "'";
                //mariadb   date

            } else if ("datetime".equalsIgnoreCase(dataType) && value instanceof Long) {
                return "'" + datetimeToStringFromUTC((Long) value) + "'";
                //mariadb   date

            } else if (("timestamp".equalsIgnoreCase(dataType)) && value instanceof Long) {
                return "'" + dateToString((Long) value) + "'";
                //mariadb   date

            } else {
                return value;
            }
        }

        private String dateToStringFromUTC(Long date) {
            DateTime dt = new DateTime(date, DateTimeZone.UTC);
            return dt.toString(DateUtil.DATE_PATTERN_ONLY_DATE);
        }

        private String datetimeToStringFromUTC(Long date) {
            DateTime dt = new DateTime(date, DateTimeZone.UTC);
            return dt.toString(DateUtil.DATE_PATTERN_FULL);
        }

        private String dateToString(Long date) {
            DateTime dt = new DateTime(date);
            return dt.toString(DateUtil.DATE_PATTERN_FULL);
        }

        private String dateToString(Date date, String dateType) {
            if ("timestamp".equalsIgnoreCase(dateType)) {
                DateTime dt = new DateTime(date);
                return dt.toString(DateUtil.DATE_PATTERN_FULL);
            } else if ("datetime".equalsIgnoreCase(dateType)) {
                DateTime dt = new DateTime(date, DateTimeZone.UTC);
                return dt.toString(DateUtil.DATE_PATTERN_FULL);
            } else if ("date".equalsIgnoreCase(dateType)) {
                DateTime dt = new DateTime(date, DateTimeZone.UTC);
                return dt.toString(DateUtil.DATE_PATTERN_ONLY_DATE);
            } else {
                DateTime dt = new DateTime(date);
                return dt.toString(DateUtil.DATE_PATTERN_FULL);
            }

        }

        private void handleUpdateRowsEvent(Event event) {
            UpdateRowsEventData eventData = event.getData();
            TableMapEventData tableMapEvent = tablesById.get(eventData.getTableId());
            if (isShouldBeFilter(tableMapEvent.getDatabase(), tableMapEvent.getTable()))
                return;
            Map<Integer, Map<String, Object>> xxx = tablesColumnMap.get(tableMapEvent.getDatabase() + "." + tableMapEvent.getTable());
            BitSet inculudeColumn = eventData.getIncludedColumns();
            StringBuilder sba = new StringBuilder("update ");
            sba.append(tableMapEvent.getTable());
            sba.append(" set ");
            int size = inculudeColumn.length();

            List<Map.Entry<Serializable[], Serializable[]>> rows = eventData.getRows();
            for (Map.Entry<Serializable[], Serializable[]> row : rows) {
                StringBuilder sb = new StringBuilder(sba);
                int slot = -1;
                Map.Entry<Serializable[], Serializable[]> rowMap = row;
                Serializable[] value = rowMap.getValue();
                Serializable[] key = rowMap.getKey();
                for (int i = 0; i < size; i++) {
                    int column = inculudeColumn.nextSetBit(i);
                    Map<String, Object> coumnMap = xxx.get(column + 1);
                    sb.append(coumnMap.get("COLUMN_NAME"));
                    sb.append("=");
                    String dataType = (String) coumnMap.get("DATA_TYPE");
                    sb.append(convertBinlogValue(value[i], dataType));

                    if (i != size - 1) {
                        sb.append(",");
                    }
                }
                sb.append(" where ");

                BitSet includedColumnsBeforeUpdate = eventData.getIncludedColumnsBeforeUpdate();
                for (int i = 0; i < size; i++) {
                    int column = includedColumnsBeforeUpdate.nextSetBit(i);
                    Map<String, Object> coumnMap = xxx.get(column + 1);
                    sb.append(coumnMap.get("COLUMN_NAME"));
                    Serializable value1 = key[i];
                    if (value1 == null) {
                        sb.append(" is null");
                    } else {
                        sb.append("=");
                        String dataType = (String) coumnMap.get("DATA_TYPE");

                        sb.append(convertBinlogValue(value1, dataType));
                    }
                    String columnName = (String) coumnMap.get("COLUMN_NAME");
                    if ("_slot".equalsIgnoreCase(columnName)) {
                        slot = value1 instanceof BigInteger ? ((BigInteger) value1).intValue() : ((Integer) value1);
                    }
                    if (i != size - 1) {
                        sb.append(" and ");
                    }
                }

                checkIfExeSql(tableMapEvent, sb, slot);
            }


        }

        private void handleDeleteRowsEvent(Event event) {
            DeleteRowsEventData eventData = event.getData();
            TableMapEventData tableMapEvent = tablesById.get(eventData.getTableId());
            if (isShouldBeFilter(tableMapEvent.getDatabase(), tableMapEvent.getTable()))
                return;
            Map<Integer, Map<String, Object>> xxx = tablesColumnMap.get(tableMapEvent.getDatabase() + "." + tableMapEvent.getTable());
            BitSet inculudeColumn = eventData.getIncludedColumns();
            StringBuilder sba = new StringBuilder("delete from ");
            sba.append(tableMapEvent.getTable());
            sba.append(" where ");
            int size = inculudeColumn.length();
            List<Serializable[]> rows = eventData.getRows();
            for (Serializable[] row : rows) {
                StringBuilder sb = new StringBuilder(sba);
                Serializable[] value = row;


                int slot = -1;
                for (int i = 0; i < size; i++) {
                    int column = inculudeColumn.nextSetBit(i);
                    Map<String, Object> coumnMap = xxx.get(column + 1);
                    sb.append(coumnMap.get("COLUMN_NAME"));
                    Serializable value1 = value[i];
                    if (value1 == null) {
                        sb.append(" is null");
                    } else {
                        sb.append("=");
                        String dataType = (String) coumnMap.get("DATA_TYPE");

                        sb.append(convertBinlogValue(value1, dataType));
                    }
                    String columnName = (String) coumnMap.get("COLUMN_NAME");
                    if ("_slot".equalsIgnoreCase(columnName)) {
                        slot = value1 instanceof BigInteger ? ((BigInteger) value1).intValue() : ((Integer) value1);
                    }
                    if (i != size - 1) {
                        sb.append(" and ");
                    }
                }
                checkIfExeSql(tableMapEvent, sb, slot);

            }


        }


    }

    public static void main(String[] args) {
        BinlogStream stream = new BinlogStream("localhost", 3306, "root", "123", Charset.defaultCharset());
        try {
            stream.setSlaveID(23511);
            stream.setBinglogFile("mysql-bin.000028");
            stream.setBinlogPos(1082);
            stream.connect();

        } catch (IOException e) {
            e.printStackTrace();
        }

//        String sql="2'aa\"啊啊402";
//        System.out.println(sql.replace("'","\\'"));
    }
}
